package com.lx.top.n;

import com.lx.entitys.CovidTopEntity;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: chenjiang
 * @Date: 2021/11/19/17:23
 * @Description: <重点理解 MapReduce中Reduce端中进行聚合操作 key和Iterable values的关系/>
 * @desc: <br>
 * 1.在本次案例中 我们想实现的是每个州的确证病例数 最多的前3县个的数据 由此引申出来 当我们数据经过Map端输出的是JavaBean对象在reduce端 如何进行统计的
 *   1.1 我们知道Map端输出结果 到Reduce端会经过 排序和分组的操作
 *   1.1.1 排序操作
 *      1.1.1由于我自己重新定义了排序规则 先对州进行排序  州相当 再根据确证病例数倒叙排序
 *      {@link CovidTopEntity#compareTo(CovidTopEntity)}
 *@Code:
 *     public int compareTo(CovidTopEntity o) {
 *         int result;
 *         int i = state.compareTo(o.getState());
 *         if (i > 0) {
 *             result = 1;
 *         } else if (i < 0) {
 *             result = -1;
 *         } else {
 *             //州相等 同一个州下面按照 确诊病例数倒序排序
 *             result = cases > o.getCases() ? -1 : 1;
 *         }
 *         return result;
 *     }
 *     当排序后的数据如下
 *     0.CovidTopEntity{state="Alabama",county="Autauga",cases=3456}
 *     1.CovidTopEntity{state="Alabama",county="Barbour",cases=123}
 *     3.CovidTopEntity{state="Arkansas",county="Crawford",cases=6093}
 *     4.CovidTopEntity{state="California",county="El Dorado",cases=234}
 *     2.CovidTopEntity{state="Arkansas",county="Baxter",cases=6899}
 *     ..................................
 *
 *   1.1.1 分组操作
 *         由于重新编写了分组操作我们按照的是当前对象的state (州) 的属性进行分组 那么相同的州就会分到同一个组
 * @code:
 *    @Override
 *     public int compare(WritableComparable a, WritableComparable b) {
 *         CovidTopEntity aBean = (CovidTopEntity) a;
 *         CovidTopEntity bBean = (CovidTopEntity) b;
 *         return aBean.getState().compareTo(bBean.getState());
 *     }
 *     {@link com.lx.top.CovidGroupingComparator#compare(WritableComparable, WritableComparable)}
 *     此时数据会被分为三个组分别为组内根据cases倒叙排序:
 *     top1:
 *          key=CovidTopEntity{state="Alabama",county="Barbour",cases=3456} value=3456
 *          key=CovidTopEntity{state="Alabama",county="Autauga",cases=123} value=123
 *     top2:
 *          key=CovidTopEntity{state="Arkansas",county="Baxter",cases=6899} value=6899
 *          key=CovidTopEntity{state="Arkansas",county="Crawford",cases=6093} value=6093
 *       此刻数据到Reduce端：每组调用进行reduce计算
 * </br>
 */
public class CovidTopNReducer extends Reducer<CovidTopEntity, LongWritable, CovidTopEntity, LongWritable> {
    @Override
    protected void reduce(CovidTopEntity key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        Long count=1L;
        for (LongWritable value : values) {
            if(count<3){
                context.write(key,value);
                count++;
            }else {
                return;
            }
        }
    }
}
